package cn.easy.rpc.proxy;

import cn.easy.rpc.annotations.RpcInstance;
import cn.easy.rpc.annotations.RpcReference;
import cn.easy.rpc.config.ConfigHolder;
import cn.easy.rpc.context.RpcContext;
import cn.easy.rpc.core.DataExchangeIdentity;
import cn.easy.rpc.core.ServiceRegistrar;
import cn.easy.rpc.trans.TransModel;
import lombok.AllArgsConstructor;
import lombok.Getter;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.cglib.proxy.MethodInterceptor;
import org.springframework.cglib.proxy.MethodProxy;
import org.springframework.util.CollectionUtils;

import java.io.*;
import java.lang.reflect.Method;
import java.net.Socket;
import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;

/**
 * @author zwb
 * @create 2020-09-18 21:49
 **/
@Getter
public class RpcInvokeInterceptor implements MethodInterceptor, Serializable {

    private static final Log logger = LogFactory.getLog(RpcInvokeInterceptor.class);


    public RpcInvokeInterceptor(RpcReference rpcReference) {
        this.rpcReference = rpcReference;
    }

    private final RpcReference rpcReference;

    private final RpcContext rpcContext = RpcContext.getInstance();
    private final ServiceRegistrar selfService = ConfigHolder.config;

    @Override
    public Object intercept(Object o, Method method, Object[] args, MethodProxy methodProxy) throws Throwable {
        ServiceRegistrar serverNode = getServerNode();
        logger.info("rpc invoke " +
                "host[" + serverNode.getHost() + "] " +
                "port[" + serverNode.getPort() + "] " +
                "appName[" + serverNode.getName() + "] " +
                "version[" + serverNode.getVersion() + "] " +
                "schema[" + rpcReference.schema() + "] " +
                "method:[" + method.getName() + "] " +
                "args:" + Arrays.asList(args).toString());
        TransModel transModel = new TransModel();
        transModel.setSchema(rpcReference.schema());
        transModel.setMethodName(method.getName());
        transModel.setArgs(args);
        return RpcInvoke(serverNode.getHost(), serverNode.getPort(), transModel);
    }

    private ServiceRegistrar getServerNode() {
        List<ServiceRegistrar> dependService = rpcContext.getDependService();
        //TODO 后续扩展的服务路由
        List<ServiceRegistrar> serverNodes = dependService.stream().filter(serverNode ->
                serverNode.getGroup().equals(selfService.getGroup()) && serverNode.getName().equals(rpcReference.appName()) && serverNode.getVersion().equals(rpcReference.version())
        ).collect(Collectors.toList());
        if (CollectionUtils.isEmpty(serverNodes))
            throw new RuntimeException("No service available serviceNode:" + rpcReference.appName() + " version:" + rpcReference.version());
        return serverNodes.stream().findFirst().orElse(null);
    }

    private Object RpcInvoke(String serverHost, Integer serverPort, TransModel transModel) {
        Socket socket = null;
        ObjectInputStream is = null;
        ObjectOutputStream os = null;
        try {
            socket = new Socket(serverHost, serverPort);
            os = new ObjectOutputStream(socket.getOutputStream());
            os.writeObject(transModel);
            is = new ObjectInputStream(new BufferedInputStream(
                    socket.getInputStream()));
            return is.readObject();
        } catch (IOException | ClassNotFoundException exception) {
            exception.printStackTrace();
            throw new RuntimeException("Ppc invoke exception serviceNode:" + rpcReference.appName() + " version:" + rpcReference.version());
        } finally {
            try {
                if (socket != null)
                    socket.close();
                if (is != null)
                    is.close();
                if (os != null)
                    os.close();
            } catch (IOException ioException) {
                ioException.printStackTrace();
            }
        }
    }
}
